package com.alison.tableapisql.chapter1_tableapiandsql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Elasticsearch;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Schema;

/**
 * @Author alison
 * @Date 2024/4/12 10:48
 * @Version 1.0
 * @Description
 */
public class E2_TableMySql {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String filePath = "D:\\workspace\\lab\\learnbigdata\\learnflink\\flink-datastream\\src\\main\\resources\\tableapi\\E1.txt";
        tableEnv.connect(new FileSystem().path(filePath))
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");

        String sinkDDL =
                "create table jdbcOutputTable (" +
                        " id STRING  , " +
                        " cnt BIGINT  ," +
                        " PRIMARY KEY (id) not ENFORCED "+
                        ") with ( " +
                        " 'connector' = 'jdbc', " +
                        " 'url' = 'jdbc:mysql://localhost:3306/flink', " +
                        " 'table-name' = 'sensor_count', " +
                        " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                        " 'username' = 'root', " +
                        " 'password' = 'root' " +
                        ")";
        tableEnv.executeSql(sinkDDL);// 执行DDL创建表

        tableEnv.executeSql("desc inputTable").print();
        tableEnv.executeSql("desc jdbcOutputTable").print();
        tableEnv.executeSql("select * from inputTable").print();
        tableEnv.executeSql("select id, temp from inputTable where id = 'sensor_6'").print();
        System.out.println("-----------------------");
        tableEnv.sqlQuery("select id, count(id) as `cnt`  from inputTable group by id").executeInsert("jdbcOutputTable");
    }
}
